/* Copyright  2009-2010
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/**
 * @filename: lockManager.cc
 */
#include<server_includes.h>
#include<lock.h>
#include<allocator.h>
#include<database.h>
#include<catalogtables.h>
#include<transaction.h>
#include<debug.h>
#include<read-config.h>
#include<process.h>

#include <gettext.h>
#include <error.h>
#include <errmsg_print.h>

using namespace BackEnd ;

LockManager::LockManager(Database *sysdb_)
{
  systemDatabase_ = sysdb_;
  lockBuckets = systemDatabase_->getLockHashBuckets();
}

LockManager::LockManager(Database *sysDb_,Database* userDb_,
                         DatabaseManagerImpl* dbmanger)
{
  systemDatabase_ = sysDb_;
  lockBuckets = systemDatabase_->getLockHashBuckets();
  userDatabase_ = userDb_;
  dbmgr_  = dbmanger;
}

///   get the Bucket.
///   The algorithm is very simple,just use the KEY % LOCK_BUCKET_SIZE
Bucket* LockManager::getLockBucket(void *tuple)
{
   ulong key =(ulong)tuple ;
   int bucketNo = key % LOCK_BUCKET_SIZE   ;
   errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] getLockBucket bucketNo is [%d]\n"),
                     __FILE__,__LINE__,bucketNo);
   return &(lockBuckets[bucketNo]);
}

void LockManager::printUsageStatistics()
{
   Bucket* buckets = systemDatabase_->getLockHashBuckets();
   Bucket* bucket;
   LockHashNode *lockNode;
   int nodeCount =0, bucketCount =0;

   for (int i =0; i< LOCK_BUCKET_SIZE; i++)
   {
       bucket = &(buckets[i]);
       lockNode = (LockHashNode*) bucket->bucketList_;
       if (lockNode) bucketCount++; else continue;
       while (NULL != lockNode) { nodeCount++; lockNode = lockNode->next_; }
   }
   printf("<LockTable>\n");
   printf("  <TotalBuckets> %d  </TotalBuckets>\n", LOCK_BUCKET_SIZE);
   printf("  <UsedBuckets> %d  </UsedBuckets>\n", bucketCount);
   printf("  <TotalLockNodes> %d  </TotalLockNodes>\n", nodeCount);
   printf("</LockTable>\n");

}

void LockManager::printDebugInfo()
{
   Bucket* buckets = systemDatabase_->getLockHashBuckets();
   Bucket* bucket;
   LockHashNode *lockNode;
   int nodeCount =0, bucketCount =0;
   printf("<LockTable>\n");
   for (int i =0; i< LOCK_BUCKET_SIZE; i++)
   {
       nodeCount =0;
       bucket = &(buckets[i]);

       lockNode = (LockHashNode*) bucket->bucketList_;

       while (NULL != lockNode)
       {
           nodeCount++;
           lockNode->print();
           lockNode = lockNode->next_;
       }
       if (nodeCount) {
           bucketCount++;
           printf("  <LockBucket> \n");
           printf("    <BucketNo> %d </BucketNo> \n", i);
           printf("    <TotalNodes> %d </TotalNodes>\n", nodeCount);
           printf("  <LockBucket>\n");
       }
   }

   printf("  <TotalUsedBuckets> %d </TotalUsedBuckets>\n", bucketCount);
   Chunk *chunk = systemDatabase_->getSystemDatabaseChunk(LockTableId);
   printf("  <TotalPages> %ld </TotalPages>\n",(long)chunk->totalPages());
   printf("</LockTable>\n");

}

DbRetVal LockManager::getSharedLock(void *tuple, Transaction **trans)
{
   //get the bucket list
   //take the bucket mutex for read
   //go the the next level bucket list
   //get the bucket iterator
   //go the node where the lock info resides
   //check which mode the lock is taken
   // if shared then
   //      upgrade the bucket mutex to write
   //      take it and increment the readers count
   //      release bucket mutex and exit
   // if exclusive then
   //       go into the loop
   //       upgrade the bucket mutex to write
   //       increment waitReaders count
   //       release the bucket mutex
   //       wait for timeout period or (takes shared lock and release it ) till it becomes free.
   //       if times out
   //            take bucket mutex for write
   //            decrement waitReaders count
   //            release bucket mutex

   //            return
   //       if it becomes free
   //            take bucket mutex for write
   //            increment readers
   //            release bucket mutex

   //            return


   LockInfo linfo;
   linfo.noOfReaders_ = 1;
   //keeping it ready for the allocation, because when
   //lock node is not present in the list, then it means we are the first
   //to acquire lock so for sure we will get it.
   Bucket *bucket = getLockBucket(tuple);

   int lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
   if (lockRet != 0)
       return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;

   LockHashNode *lockNode = (LockHashNode*) bucket->bucketList_;
   if (NULL == lockNode)
   {
        DbRetVal rv = OK;
        LockHashNode *node = allocLockNode(linfo, tuple, &rv);
        if (NULL == node)return rv;

        bucket->bucketList_ = (void*)node; //make it as head
        bucket->mutex_.releaseLock(systemDatabase_->procSlot);
        rv = OK;
        if (trans != NULL)
            rv = (*trans)->insertIntoHashList(systemDatabase_, node);
        if (rv !=OK) linfo.noOfReaders_--;
        return rv;
   }
   LockHashNode *cachedLockNode = NULL;

   LockHashNode *iter = lockNode;
   //Iterate though the list and find the element's lock info
   while(iter != NULL)
   {
       if(iter->ptrToTuple_ == tuple)
       {
           if (iter->lInfo_.noOfReaders_ == -1)
           {

               iter->lInfo_.waitReaders_++;
                       cachedLockNode = iter;
               bucket->mutex_.releaseLock(systemDatabase_->procSlot);
               if (trans != NULL) (*trans)->updateWaitLock(iter);
               break;
           }
           else if (iter->lInfo_.noOfReaders_ == 0)
           {
               if(iter->lInfo_.waitWriters_ >0)
               {
                   iter->lInfo_.waitReaders_++;
                   cachedLockNode = iter;
                   bucket->mutex_.releaseLock(systemDatabase_->procSlot);
                   if (trans != NULL) (*trans)->updateWaitLock(iter);
                   break;
               }
               else
               {
                   iter->lInfo_.noOfReaders_++;
                   bucket->mutex_.releaseLock(systemDatabase_->procSlot);
                   DbRetVal rv = OK;
                   if (trans != NULL)
                     rv = (*trans)->insertIntoHashList(systemDatabase_, iter);
                   if (rv != OK) iter->lInfo_.noOfReaders_--;
                   return rv;
                }
           }else
           {
               iter->lInfo_.noOfReaders_++;
               bucket->mutex_.releaseLock(systemDatabase_->procSlot);
               DbRetVal rv = OK;
               if (trans != NULL)
                 rv = (*trans)->insertIntoHashList(systemDatabase_, iter);
               if (rv != OK) iter->lInfo_.noOfReaders_--;
               return rv;
           }
       }
       iter = iter->next_;
   }
   if (NULL == cachedLockNode)
   {
        DbRetVal rv =OK;
        LockHashNode *node = allocLockNode(linfo, tuple, &rv);
        if (NULL == node)
        {
            bucket->mutex_.releaseLock(systemDatabase_->procSlot);
            if (trans != NULL) (*trans)->removeWaitLock();
            return rv;
        }

        LockHashNode *it = lockNode;
        while (NULL != it->next_) it = it->next_;
        it->next_ = node;
        bucket->mutex_.releaseLock(systemDatabase_->procSlot);
        rv = OK;
        if (trans != NULL)
          rv = (*trans)->insertIntoHashList(systemDatabase_, node);
        if (rv != OK) linfo.noOfReaders_--;
        if (trans != NULL) (*trans)->removeWaitLock();
        return rv;
   }

   int tries = 0;

   struct timeval timeout;
   timeout.tv_sec = Conf::config.getLockSecs();
   timeout.tv_usec = Conf::config.getLockUSecs();

   while (tries < Conf::config.getLockRetries())
   {
       lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
       if (lockRet != 0)
       {
           if (trans != NULL) (*trans)->removeWaitLock();
           return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
       }
       if (cachedLockNode->lInfo_.noOfReaders_ == 0)
       {
           //if there are waiters allow then to take the lock
           if (cachedLockNode->lInfo_.waitWriters_ <0)
           {
               cachedLockNode->lInfo_.noOfReaders_++;
               cachedLockNode->lInfo_.waitReaders_--;
               bucket->mutex_.releaseLock(systemDatabase_->procSlot);
               DbRetVal rv = OK;
               if (trans != NULL)
                 rv = (*trans)->insertIntoHashList(systemDatabase_,
                                                  cachedLockNode);
               if (rv !=OK) {
                   cachedLockNode->lInfo_.noOfReaders_--;
                   cachedLockNode->lInfo_.waitReaders_++;
                   if (trans != NULL) (*trans)->removeWaitLock();
                   return rv;
               }
               if (trans != NULL) (*trans)->removeWaitLock();
               return OK;
           }
       } else if (cachedLockNode->lInfo_.noOfReaders_ == -1)
       {
           if (trans !=NULL && (*trans)->findInHashList(systemDatabase_,
                                                       cachedLockNode))
           {
               bucket->mutex_.releaseLock(systemDatabase_->procSlot);
               if (trans != NULL) (*trans)->removeWaitLock();
               return OK;
           }
       } else
       {
           cachedLockNode->lInfo_.noOfReaders_++;
           cachedLockNode->lInfo_.waitReaders_--;
           bucket->mutex_.releaseLock(systemDatabase_->procSlot);
           DbRetVal rv =OK;
           if (trans != NULL)
              rv = (*trans)->insertIntoHashList(systemDatabase_, cachedLockNode);
           if (rv !=OK) {
               cachedLockNode->lInfo_.noOfReaders_--;
               cachedLockNode->lInfo_.waitReaders_++;
           }
           if (trans != NULL) (*trans)->removeWaitLock();
           return rv;
       }

       bucket->mutex_.releaseLock(systemDatabase_->procSlot);
       OS::select(0, 0, 0, 0, &timeout);
       tries++;
   }
   if (trans != NULL) (*trans)->removeWaitLock();
   return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;

}

DbRetVal LockManager::getExclusiveLock(void *tuple, Transaction **trans)
{
   LockInfo linfo;
   linfo.noOfReaders_ = -1;
   Session* _session = current_session ;
   errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] getExclusiveLock Begin: tuple [%p] trans [%p]\n"),
                     __FILE__,__LINE__,tuple, trans);
   int _procslot = 0;
   if(current_session){_procslot=current_session->getProcSlot();}
   //keeping it ready for the allocation, because when
   //lock node is not present in the list, then it means we are the first
   //to acquire lock so for sure we will get it.

   Bucket *bucket = getLockBucket(tuple);
   int lockRet = bucket->mutex_.getLock(_procslot);
   if (lockRet != 0)
   {
       push_warning(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                        ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
       return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
   }
   BKT_TREE_ARG custom_arg;

   custom_arg.search_flag= SEARCH_FIND | SEARCH_UPDATE;
   custom_arg.key_length= sizeof(void*);
   custom_arg.dbptr = userDatabase_;
   custom_arg.chunkptr = bucket->element_chunk;

   if (OK !=userDatabase_->getDatabaseMutex()){
     bucket->mutex_.releaseLock(_procslot);
     return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
   }
   int ret ;
   BUCKET_TREE_ELEMENT* the_element = bucket->subtree.tree_insert(
                       (void*)tuple,custom_arg.key_length, &custom_arg,ret);
   userDatabase_->releaseDatabaseMutex();
   if(!the_element)
   {
     bucket->mutex_.releaseLock(_procslot);
     return (DbRetVal)ret;
   }
   LockHashNode *lockNode = (LockHashNode*) the_element->hashbucketList_;
   /// if  lockNode is NULL.
   if (!lockNode)
   {
        DbRetVal rv = OK;

        LockHashNode *node = allocLockNode(linfo, tuple, &rv);
        if (!node)
        {
         bucket->mutex_.releaseLock(_procslot);
          return rv;
        }
        the_element->hashbucketList_ = (void*)node; //make it as head
        bucket->mutex_.releaseLock(_procslot);
        rv =OK;
        if (trans != NULL)
           rv = (*trans)->insertIntoHashList(systemDatabase_, node);
        if (rv !=OK) node->lInfo_.noOfReaders_ = 0;
        return rv;
   }
   LockHashNode *cachedLockNode = lockNode;

   if(cachedLockNode->lInfo_.noOfReaders_ !=0)
   {
      cachedLockNode->lInfo_.waitWriters_++;
      bucket->mutex_.releaseLock(_procslot);
      if (trans != NULL) (*trans)->updateWaitLock(cachedLockNode);
      goto START_TO_WAIT;
   }else{
      cachedLockNode->lInfo_.noOfReaders_ = -1;
      bucket->mutex_.releaseLock(_procslot);
      DbRetVal rv =OK;
      if (trans != NULL)
      rv = (*trans)->insertIntoHashList(systemDatabase_, cachedLockNode);
      if (rv != OK) cachedLockNode->lInfo_.noOfReaders_ = 0;
      return rv;
   }

START_TO_WAIT:
   int tries = 0;

   struct timeval timeout;
   timeout.tv_sec = Conf::config.getLockSecs();
   timeout.tv_usec = Conf::config.getLockUSecs();

   while (tries < Conf::config.getLockRetries())
   {
       lockRet = bucket->mutex_.getLock(_procslot);
       if (lockRet != 0)
       {
         return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;;
       }
       if (cachedLockNode->lInfo_.noOfReaders_ == 0)
       {
           cachedLockNode->lInfo_.noOfReaders_ = -1;
           cachedLockNode->lInfo_.waitWriters_--;
           bucket->mutex_.releaseLock(_procslot);
           DbRetVal rv = OK ;
           if (trans != NULL)
               rv = (*trans)->insertIntoHashList(systemDatabase_, cachedLockNode);
           if (trans != NULL) (*trans)->removeWaitLock();
           return rv;
       }else if ( cachedLockNode->lInfo_.noOfReaders_ == 1)
       {
           if (trans !=NULL &&
              (*trans)->findInHashList(systemDatabase_, cachedLockNode))
           {
               //upgrade it to exclusive lock
               cachedLockNode->lInfo_.noOfReaders_ = -1;
               cachedLockNode->lInfo_.waitWriters_--;
               bucket->mutex_.releaseLock(_procslot);
               if (trans != NULL) (*trans)->removeWaitLock();
               return OK;
           }
           if (trans ==NULL
            && ProcessManager::hasLockList.exists(cachedLockNode->ptrToTuple_))
           {
               //upgrade it to exclusive lock
               cachedLockNode->lInfo_.noOfReaders_ = -1;
               cachedLockNode->lInfo_.waitWriters_--;
               bucket->mutex_.releaseLock(_procslot);
               if (trans != NULL) (*trans)->removeWaitLock();
               return OK;
           }
       }else if ( cachedLockNode->lInfo_.noOfReaders_ == -1)
       {
           if (trans !=NULL
               && (*trans)->findInHashList(systemDatabase_, cachedLockNode))
           {
               cachedLockNode->lInfo_.waitWriters_--;
               bucket->mutex_.releaseLock(_procslot);
               if (trans != NULL) (*trans)->removeWaitLock();
               return OK;
           }
           if (trans ==NULL
            && ProcessManager::hasLockList.exists(cachedLockNode->ptrToTuple_))
           {
               cachedLockNode->lInfo_.waitWriters_--;
               bucket->mutex_.releaseLock(_procslot);
               if (trans != NULL) (*trans)->removeWaitLock();
               return OK;
           }
       }
       bucket->mutex_.releaseLock(_procslot);
       OS::select(0, 0, 0, 0, &timeout);
       tries++;
   }

   if (trans != NULL) (*trans)->removeWaitLock();

   return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
}

DbRetVal LockManager::releaseLock(void *tuple)
{
   LockInfo linfo;
   linfo.noOfReaders_ = 0;
   //keeping it ready for the allocation, because when
   //lock node is not present in the list, then it means we are the first
   //to acquire lock so for sure we will get it.
   //Session* _session = current_session ;
   errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] releaseLock Begin: tuple [%p]\n"),
                     __FILE__,__LINE__,tuple);
   Bucket *bucket = getLockBucket(tuple);
   int _procslot = 0;
   if(current_session){_procslot=current_session->getProcSlot();}
   int lockRet = bucket->mutex_.getLock(_procslot);
   if (lockRet != 0)
       return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
   BKT_TREE_ARG custom_arg;

   custom_arg.search_flag= SEARCH_FIND | SEARCH_UPDATE;
   custom_arg.key_length= sizeof(void*);
   custom_arg.dbptr = userDatabase_;
   custom_arg.chunkptr = bucket->element_chunk;
   custom_arg.search_type = get_element;
   BUCKET_TREE_ELEMENT* the_element =(BUCKET_TREE_ELEMENT* ) bucket->subtree.tree_search(
                          (void*)tuple,&custom_arg);
   if(!the_element)
   {
     bucket->mutex_.releaseLock(_procslot);
     return (DbRetVal)HA_ERR_SYS_INTERNAL;
   }
   LockHashNode *lockNode = (LockHashNode*) the_element->hashbucketList_;
   if (!lockNode)
   {
       bucket->mutex_.releaseLock(_procslot);
       return (DbRetVal)HA_ERR_SYS_INTERNAL;
   }

   if(lockNode->lInfo_.noOfReaders_ == -1) {
     if (lockNode->lInfo_.waitWriters_ == 0 || lockNode->lInfo_.waitReaders_ ==0)
     {
       deallocLockNode(lockNode);
       userDatabase_->getDatabaseMutex();
       bucket->subtree.tree_delete((void*)tuple,custom_arg.key_length,&custom_arg);
       userDatabase_->releaseDatabaseMutex();
       goto RELEASE_LOCK;
     }else{
       goto RELEASE_LOCK;
     }
   }
   else if (lockNode->lInfo_.noOfReaders_ == 1)
   {
    if (lockNode->lInfo_.waitWriters_ == 0 || lockNode->lInfo_.waitReaders_ ==0)
    {
      deallocLockNode(lockNode);
      userDatabase_->getDatabaseMutex();
      bucket->subtree.tree_delete((void*)tuple,custom_arg.key_length,&custom_arg);
      userDatabase_->releaseDatabaseMutex();
      goto RELEASE_LOCK;
    }else{
      lockNode->lInfo_.noOfReaders_--;
      goto RELEASE_LOCK;
    }
   }else
   {
      lockNode->lInfo_.noOfReaders_--;
      goto RELEASE_LOCK;
   }


RELEASE_LOCK:
   bucket->mutex_.releaseLock(_procslot);
   return OK;
}

DbRetVal LockManager::isExclusiveLocked(void *tuple, Transaction **trans, bool &status)
{
   Bucket *bucket = getLockBucket(tuple);
   int _procslot = 0;
   if(current_session){_procslot=current_session->getProcSlot();}
   int lockRet = bucket->mutex_.getLock(_procslot);
   if (lockRet != 0)
      return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;

   BKT_TREE_ARG custom_arg;

   custom_arg.search_flag= SEARCH_FIND | SEARCH_UPDATE;
   custom_arg.key_length= sizeof(void*);
   custom_arg.dbptr = userDatabase_;
   custom_arg.chunkptr = bucket->element_chunk;
   custom_arg.search_type = get_element;
   BUCKET_TREE_ELEMENT* the_element =(BUCKET_TREE_ELEMENT* ) bucket->subtree.tree_search(
                          (void*)tuple,&custom_arg);
   if(!the_element)
   {
     status = false;
     bucket->mutex_.releaseLock(_procslot);
     return OK;
   }

   LockHashNode *lockNode = (LockHashNode*)the_element->hashbucketList_;
   LockHashNode *iter = lockNode;
   //Iterate though the list and find the element's lock info
   //Only exclusive locks are checked. shared locks are not considered for this
   if(iter != NULL)
   {

         if (iter->lInfo_.noOfReaders_ == -1)
       {
          if (trans != NULL && (*trans)->findInHashList(systemDatabase_, iter))
         {
           status = false;
         }else
           status = true;
         bucket->mutex_.releaseLock(_procslot);
         return OK;
      }
   }
   status = false;
   bucket->mutex_.releaseLock(_procslot);

   return OK;
}
///   Allocate Lock Node.
///
///
LockHashNode* LockManager::allocLockNode(LockInfo &info, void *tuple,
                                         DbRetVal *rv)
{
    //allocate lock node
    /// chunk is pointing to the Chunk (LockTableId);
    Chunk *chunk = systemDatabase_->getSystemDatabaseChunk(LockTableId);

    LockHashNode *node = (LockHashNode*)chunk->allocate(systemDatabase_, rv);

    if (!node)
      return NULL;

    node->ptrToTuple_ = tuple;
    node->lInfo_      = info;
    node->next_       = NULL;
    return node;
}

void LockManager::deallocLockNode(LockHashNode *node, Bucket *bucket)
{
    Chunk *chunk = systemDatabase_->getSystemDatabaseChunk(LockTableId);

    LockHashNode *nodeList = (LockHashNode*) bucket->bucketList_;

    LockHashNode *iter = nodeList, *prev = nodeList;
    if (NULL == nodeList)
      return;

    //If it is the first node, then make the bucket point to the next node
    //in the list
    if (nodeList == node)
    {
       bucket->bucketList_ = node->next_;
       chunk->free(systemDatabase_, node);
       return;
    }
    while(iter != node)
    {
        prev = iter;
        iter = iter->next_;
    }
    //delete the node by making previous element point to the next element
    //of the deleted element in the list
    prev->next_ = iter->next_;

    chunk->free(systemDatabase_, node);
    return ;
}

void LockManager::deallocLockNode(LockHashNode *node)
{
    Chunk *chunk = systemDatabase_->getSystemDatabaseChunk(LockTableId);
    chunk->free(systemDatabase_, node);
}
